import sys
import subprocess
import json
import getopt

mapper_cmd = None
reducer_cmd = None
input_path = None

try:
    # 解析命令行参数
    opts, args = getopt.getopt(sys.argv[1:], "m:r:h", ["mapper=", "reducer=", "help"])
    for opt, val in opts:
        if opt in ("-m", "--mapper"):
            mapper_cmd = val
        elif opt in ("-r", "--reducer"):
            reducer_cmd = val
        elif opt in ("-h",):
            print("usage: %s --mapper MAP_CMD --reducer REDUCE_CMD INPUT_PATH")

    if not args:
        raise Exception("not input path specified")

    input_path = args[0]
    if not ( mapper_cmd and reducer_cmd ):
        raise Exception("missing options --mapper(-m) and --reducer(-r)")

except Exception as e:
    print(str(e))
    sys.exit(1)


# 调用外部程序
def call_external_app(app_name, data):
    p = subprocess.Popen([app_name], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
    out, _ = p.communicate(data)
    # 处理外部程序的输出，以行为单位，每行代表一个键值对（\t分割）
    for line in out.decode("utf-8").split("\n"):
        if not line:
            continue

        try:
            key, value = line.split("\t")
        except:
            raise Exception("The key value must be separated with TAB character")

        # 反序列化键与值
        key = json.loads(key)
        value = json.loads(value)

        yield key, value


def group_key_value_pairs(itr):
    group_dict = {}
    for key, value in itr:
        if key not in group_dict:
            group_dict[key] = []

        group_dict[key].append(value)

    for key, values in group_dict.items():
        yield key, tuple(values)


with open(input_path, "r") as fp:
    data = fp.read()

reducer_data = []
# map阶段
for key, values in group_key_value_pairs(call_external_app(mapper_cmd, data.encode("utf-8"))):
    line = "{}\t{}".format(json.dumps(key), json.dumps(values))
    reducer_data.append(line)

# reduce阶段
for key, value in call_external_app(reducer_cmd, '\n'.join(reducer_data).encode("utf-8")):
    print(key, value)